package cn.com.itg.pojo.kafka;

import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;

public class KafkaCDCAssignerWithPeriodicWatermarks implements AssignerWithPeriodicWatermarks<KafkaConsumerRecord> {

    private long currentMaxTs = Long.MIN_VALUE;
    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        Watermark watermark = new Watermark(currentMaxTs);
        return watermark;
    }

    @Override
    public long extractTimestamp(KafkaConsumerRecord record, long l) {
        //获取当前记录的时间戳
        long recordTimeStamp = record.getTimeStamp();
        //更新最大时间戳
        currentMaxTs = Math.max(recordTimeStamp,currentMaxTs);
        //返回当前记录时间戳
        return recordTimeStamp;
    }
}
